package com.yanfei.zero.apis;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalTime;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import com.yanfei.zero.application.handler.secret.IgnoreDecrypt;
import com.yanfei.zero.domain.user.model.User1;

import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@Tag(name="事件源")
@Slf4j
public class EventSourceController {
	
	private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
	

	/**
	 * @throws IOException
	 */
	@Operation(summary = "事件源持续推送")
    @GetMapping(value="/eventSource",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	// @AddLock("lock:121")
//    @AddLog
	@IgnoreDecrypt
	@CrossOrigin(originPatterns = "*", methods = {RequestMethod.GET, RequestMethod.POST})
	Flux<Object>  eventSource(String id) throws IOException {
		return Flux.interval(Duration.ZERO,Duration.ofSeconds(1)).map(i -> 
		ServerSentEvent.<String>builder()
                .event("periodic-event")
                .id(String.valueOf(i))
                .data("SSE - " + LocalTime.now().toString())
                .build());
    }
	@Operation(summary = "返回单独的值")
	// @AddLock("lock:121")
//    @AddLog
	@IgnoreDecrypt
    @PostMapping(value = "/singleValue")
    public Mono<String> singleASyn(@RequestBody User1 user) {
        return Mono.just("Hello, World!");
    }
	
	
	
	
	@Operation(summary = "事件源订阅分次发布")
    @GetMapping(value="/subscribe",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	// @AddLock("lock:121")
//    @AddLog
	@IgnoreDecrypt
	@CrossOrigin(originPatterns = "*", methods = {RequestMethod.GET, RequestMethod.POST})
	SseEmitter  subscribe(String id) throws IOException {
	    
		// 但是如果设置SseEmitter存活时间为永久（参数填0），就会导致服务端无法感知客户端下线，从而使服务端维持连接池会越来越大无法释放。

        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter == null) {
        	sseEmitter = new SseEmitter();
        }
        SseEmitter sseEmitter2 = sseEmitter;
		// 必须使用新线程去做推送的功能，需要先返回SseEmitter对象给前端

	    new Thread(()->{
	    	System.out.println("发送！！！");
			try {
				Thread.sleep(1000);
				sseEmitter2.send(SseEmitter.event().name("msg").reconnectTime(2000).data("每2秒推送一次"));
			} catch (Exception e) {
				 log.error("SEE推送消息失败, 失败消息: {}", e);
			}
			sseEmitter2.onCompletion(() -> {
			        System.out.println("完成！！！");
			        }        
			        );
	    	  
	    }).start();
        
	    System.out.println("进来！！！");
		
    	return sseEmitter;
    }
    @GetMapping(path = "push")
    public String push(String id, String content) {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
	        String data = "{\"message\": \"Event " + new Date() + "\"}";
            try {
				sseEmitter.send(SseEmitter.event().name("msg").reconnectTime(1000).data("后端发送消息：" + content));
			} catch (IOException e) {
                log.error("SEE推送消息失败, 失败消息: {}", data, e);
                sseEmitter.completeWithError(e);
			}
            System.out.println("push " + id);
        }
        return "success";
    }
    @GetMapping(path = "over")
    public String over(String id) {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            sseEmitter.complete();
            sseCache.remove(id);
        }
        return "over";
    }
    
    @GetMapping(value="/events",produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @CrossOrigin(originPatterns = "*", methods = {RequestMethod.GET, RequestMethod.POST})
    public SseEmitter getEvents() {
        SseEmitter emitter = new SseEmitter();
        
        // 在另一个线程中发送数据到客户端
        Executors.newSingleThreadExecutor().execute(() -> {
            try {
                int i = 0;
                while(true) {
                	i++;
                    // 模拟实时数据更新	
                    emitter.send(SseEmitter.event().id(i+"").data("Event " + i));
                    Thread.sleep(1000);
                }
//                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }
}
